/**
 * @file str_utils.c
 * @brief Pthread functions used to deal with pthreads.
 * @version 1.0
 * @date Dec-2011
 * @author Leonardo Lino 2101101
 * @author Joaquim Ferreira 2101104
 */

#include <ctype.h>
#include <dirent.h>
#include <errno.h>
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <unistd.h>

#include "debug.h"
#include "files.h"
#include "hashtables.h"
#include "listas.h"
#include "memory.h"
#include "pthrds.h"
#include "str_utils.h"
#include "toperrors.h"
#include "spell.h"


/** @brief ERR_IO 1 - A constant that means that an error occurred.*/
#define ERR_IO 1
/** @brief Constants to use with pthreads, mutex and condition.*/
#define C_ERR_PTHREAD_CREATE        1
#define C_ERR_PTHREAD_JOIN          2
#define C_ERR_MUTEX_INIT            3
#define C_ERR_MUTEX_DESTROY         4
#define C_ERR_CONDITION_INIT        5
#define C_ERR_CONDITION_DESTROY     6
#define C_ERR_MUTEX_LOCK            7
#define C_ERR_MUTEX_UNLOCK          8
#define C_ERR_CONDITION_BROADCAST   9
/** @brief FREQ_OPTION 2 - A constant to use for define a type. */
#define FREQ_OPTION 2
/** @brief SPELL_OPTION 3 - A constant to use for define a type. */
#define SPELL_OPTION 3
/** @brief TOPERR_OPTION 4 - A constant to use for define a type. */
#define TOPERR_OPTION 4

/** @brief Function that manage the process of the buffer, the consumer threads, join the local hashtable and show the word histogram.*/
void m_thread (char *folder, int max_thr, int type, char *dictionary) 
{
  pthread_t *t_consum;
  PARAM_T param;
  int counter = 0, i;
  HASHTABLE_T **ret_hasht;
    
  t_consum = (pthread_t *) MY_MALLOC (sizeof (pthread_t) * max_thr);
  ret_hasht = (HASHTABLE_T **) MY_MALLOC (sizeof (HASHTABLE_T *) * max_thr);
  
  /**< @brief Start the mutex.*/
  if ((errno = pthread_mutex_init(&param.mutex, NULL)) != 0)
    ERROR (C_ERR_MUTEX_INIT, "Pthread_mutex_init() failed!");

  /**< @brief Start the condition variable.*/
  if ((errno = pthread_cond_init(&param.cond, NULL)) != 0)
    ERROR (C_ERR_CONDITION_INIT, "Pthread_cond_init() failed!");

  /**< @brief Reset the remaining parameters to pass to the threads.*/
  param.buffer = (char **) MY_MALLOC (sizeof (char *) * max_thr);
  memset(param.buffer, 0, sizeof(param.buffer)); /**< @brief Fill the memory with a constant byte.*/
  param.max = max_thr;
  param.stop = 0;
  param.total = 0;
  param.index_read = 0;
  param.index_write = 0;
  
    
   /**< @brief Load dictionary.*/
  if (type==TOPERR_OPTION)
	param.dictHolder= dictLoader(dictionary);
  
  if (type==FREQ_OPTION) param.type = FREQ_OPTION;
  if (type==TOPERR_OPTION) param.type = TOPERR_OPTION;
  

  /**< @brief Create the threads to run the consumers.*/
  for (i = 0; i < max_thr; i++){
    if ((errno = pthread_create(&t_consum[i], NULL, consumer, &param)) != 0)
      ERROR (C_ERR_PTHREAD_CREATE, "Pthread_create() failed!");
  }
  
  /**< @brief Start the production.*/
  process_folder(&param, folder, &counter);
  
  while (!param.stop) {
    /**< @brief If the buffer is empty or the signal is set then set the stop flag.*/
    if (!param.total || !G_continue) {
      if ((errno = pthread_mutex_lock(&(param.mutex))) != 0)
          ERROR (C_ERR_MUTEX_LOCK, "Pthread_mutex_lock() failed\n");
    
      param.stop = 1;
    
      if ((errno = pthread_cond_broadcast(&(param.cond))) != 0)
        ERROR (C_ERR_CONDITION_BROADCAST, "Pthread_cond_broadcast() failed");
      
      if ((errno = pthread_mutex_unlock(&(param.mutex))) != 0)
          ERROR (C_ERR_MUTEX_UNLOCK, "Pthread_mutex_unlock() failed\n");
      
      #ifdef SHOW_DEBUG
      DEBUG ("Producer send STOP!\n");
      #endif
    }
  }
  #ifdef SHOW_DEBUG
  DEBUG ("Number of processed text files: %d\n", counter);
  #endif
  
  /**< @brief Wait for all the consumers threads to finish.*/
  for (i = 0; i < max_thr; i++)
    if ((errno = pthread_join(t_consum[i], (void **) &ret_hasht[i])) != 0)
      ERROR(C_ERR_PTHREAD_JOIN, "Pthread_join() failed!");
  
    /**< @brief Destroy the mutex.*/
    if ((errno = pthread_mutex_destroy(&param.mutex)) != 0)
      ERROR (C_ERR_MUTEX_DESTROY, "Pthread_mutex_destroy() failed!");
        
    /**< @brief Destroy the condition.*/
    if ((errno = pthread_cond_destroy(&param.cond)) != 0)
      ERROR (C_ERR_CONDITION_DESTROY, "Pthread_cond_destroy() failed!");
    
    if (type==FREQ_OPTION){
		if (max_thr == 1){
		  convert_show_histogram (*ret_hasht);
		}else{
		  join_show_hasht(ret_hasht, max_thr);
		}
	}
	
	if (type==TOPERR_OPTION){
		outputTopHash(finalResults);
	}
	
	
	/**< @brief Final process where all memory should be released.*/
	if (type==TOPERR_OPTION){
		freeDictHolder(param.dictHolder);
		tabela_destruir(&param.dictHolder);
		freeFinalResults(finalResults);
	}
    MY_FREE (t_consum);
    MY_FREE (param.buffer);
    MY_FREE (ret_hasht);
}

/** @brief Function of the producer processing a folder and sub-folders.*/
void process_folder(PARAM_T *param, char *folder, int *counter)
{
  DIR *dir = NULL;
  struct dirent item, *result = NULL;
  struct stat stats;
  char *proc_folder = NULL, *full_path = NULL;
  LISTA_GENERICA_T *folder_llist = NULL;
  
  folder_llist = lista_criar (free);
  lista_inserir_inicio (folder_llist, folder);
  
  while (lista_numero_elementos (folder_llist) && G_continue) {  
    proc_folder = lista_remover_inicio (folder_llist);
    //#ifdef SHOW_DEBUG
    //DEBUG ("Search folder: %s\n", proc_folder);
    //#endif
    
    /**< @brief opendir: Opens a directory stream corresponding to the directory name (folder) 
    * and returns a pointer to the directory stream.*/
    if((dir = opendir(proc_folder)) == NULL)
      ERROR(ERR_IO,"Opendir error!");
        
                while(G_continue) {
                  /**< @brief readdir_r: It reads the next directory entry from the directory stream dir, 
                   * and returns it in the caller-allocated buffer pointed to by *item. 
                   * A pointer to the returned item is placed in *result.*/
                  if(readdir_r(dir,&item,&result) != 0) 
                        ERROR(ERR_IO,"Readdir Error!");
                        
                  /**< @brief It's the end of the directory stream.*/
                  if(result == NULL)
                        break;
                        
                  /**< @brief Skip this ones.*/
                  if(strcmp(item.d_name,".") == 0 || strcmp(item.d_name,"..") == 0)
                        continue;
                  
                  /**< @brief Get the full path.*/
                  full_path = path_cat(proc_folder,item.d_name);
                
                  /**< @brief lstat: get file status, return information about a file.*/
                  if(lstat(full_path,&stats) != 0)
                        ERROR(ERR_IO,"Stat error on file %s!", full_path);
                        
                  /**< @brief S_ISDIR <=> directory.*/
                  if(S_ISDIR(stats.st_mode)) {
                        lista_inserir_inicio(folder_llist, full_path);
                        /**< @brief S_IFREG <=> regular file.*/
                          }else if(S_ISREG(stats.st_mode)) {
                                        if (check_file_type (full_path, ".txt")) {
                                                put_buffer (param, full_path);
                                                (*counter)++;
                                        } else {
                                                MY_FREE (full_path);
                                        }
                        }
                }
    closedir (dir);
    MY_FREE (proc_folder);  
  }
  lista_destruir(&folder_llist);
}

/** @brief Function of the producer putting an element to the buffer.*/
void *put_buffer (PARAM_T *p, char *path_file)
{   
    if ((errno = pthread_mutex_lock(&(p->mutex))) != 0) {
      WARNING("Pthread_mutex_lock() failed\n");
      return NULL;
    }
    
    /**< @brief Waits for the buffer to have available space.*/
    while (p->total == p->max) {
      /**< @brief IF the signal is set then leave the loop.*/
      if (!G_continue)
        break;
      
      if ((errno = pthread_cond_wait(&(p->cond), &(p->mutex))) != 0) {
        WARNING("Pthread_cond_wait() failed");
        return NULL;
      }
    }
    
    /**< @brief IF the signal is set then free the element to the buffer and unlock the mutex.*/
    if (!G_continue) {
      MY_FREE (path_file);
      /**< @brief Leaving the critical section.*/       
      if ((errno = pthread_mutex_unlock(&(p->mutex))) != 0) {
        WARNING("Pthread_mutex_unlock() failed");
        return NULL;
      }
    }
      
    /**< @brief IF the signal is set then leave the function.*/
    if (!G_continue)
      return NULL;
    
    /**< @brief Put the string into the buffer.*/
    p->buffer[p->index_write] = path_file;
    #ifdef SHOW_DEBUG
    DEBUG ("Producer >> %s\n", p->buffer[p->index_write]);
    #endif
    p->index_write = (p->index_write + 1) % p->max;
    p->total++;
    #ifdef SHOW_DEBUG
    DEBUG("Total: %d\n", p->total);
    #endif
    
    /**< @brief Notify the waiting consumers.*/
    if (p->total == 1)
      if ((errno = pthread_cond_broadcast(&(p->cond))) != 0) {
        WARNING("Pthread_cond_broadcast() failed");
        return NULL;
      }
      
    /**< @brief Leaving the critical section.*/ 
    if ((errno = pthread_mutex_unlock(&(p->mutex))) != 0) {
      WARNING("Pthread_mutex_unlock() failed");
      return NULL;
    }
    return NULL;
}

/** @brief Function of the consumer(s) thread(s).*/
void *consumer (void *arg) 
{
  PARAM_T *p = (PARAM_T *) arg;
  char *filename = NULL;
  LISTA_GENERICA_T* fileholder;
  HASHTABLE_T *loc_hasht = NULL;
  
  
       if (p->type==FREQ_OPTION){
          if (p->max == 1)
                loc_hasht = tabela_criar (101, (LIBERTAR_FUNC) free_freq_word);
          else
                loc_hasht = tabela_criar (101, NULL);
       }
       
       if (p->type==TOPERR_OPTION){
		   loc_hasht = tabela_criar (101, NULL);
	   }
       

  while (1) {
    if ((errno = pthread_mutex_lock(&(p->mutex))) != 0) {
      WARNING("Pthread_mutex_lock() failed\n");
      return NULL;
    }
    
    /**< @brief Waits for the buffer to have some data.*/
    while (p->total == 0) {
      /**< @brief Check to see if the stop flag is set, if it is then exit the loop.*/
      if (p->stop)
        break;
    
      if ((errno = pthread_cond_wait(&(p->cond), &(p->mutex))) != 0) {
        WARNING("Pthread_cond_wait() failed");
        return NULL;
      }
    }
    
    /**< @brief If the stop flag is set then unlock th mutex.*/
    if (p->stop)
        if ((errno = pthread_mutex_unlock(&(p->mutex))) != 0) {
          WARNING("Pthread_mutex_unlock() failed");
          return NULL;
        }
        
    /**< @brief If the stop flag is set then exit the loop.*/
    if (p->stop)
        break;
        
    /**< @brief Retrieve one value from the buffer.*/
    #ifdef SHOW_DEBUG
    DEBUG ("Consumer Thread = [%lu] << %s\n", (unsigned long) pthread_self(), p->buffer[p->index_read]);
    #endif
    filename = p->buffer[p->index_read];
    p->index_read = (p->index_read + 1) % p->max;
    p->total--;
    #ifdef SHOW_DEBUG
    DEBUG ("Total: %d\n", p->total);
    #endif
        
    /**< @brief Notify the waiting producer.*/
    if (p->total == p->max - 1)
      if ((errno = pthread_cond_broadcast(&(p->cond))) != 0) {
        WARNING("Pthread_cond_broadcast() failed");
        return NULL;
      }
      
    /**< @brief Leaving the critical section.*/ 
    if ((errno = pthread_mutex_unlock(&(p->mutex))) != 0) {
      WARNING("Pthread_mutex_unlock(111) failed");
      return NULL;
    }
      
    printf ("Processing file: %s\n", filename);
    if (p->type==FREQ_OPTION)
                loc_hasht = getLinesOrWords (filename, loc_hasht, FREQ_OPTION);
                
    if (p->type==TOPERR_OPTION){
          fileholder=getLinesOrWords (filename, NULL, SPELL_OPTION);
          loc_hasht = processEachWordTopError(fileholder,p->dictHolder);
			

			if ((errno = pthread_mutex_lock(&(p->mutex))) != 0) {
				WARNING("Pthread_mutex_lock() failed\n");
			}
			
			mergeHashes(loc_hasht,finalResults);
			lista_destruir(&fileholder);
			
			if ((errno = pthread_mutex_unlock(&(p->mutex))) != 0) {
				WARNING("Pthread_mutex_unlock() failed");
			}
			
    }
      
    MY_FREE (filename);
  }
  
  #ifdef SHOW_DEBUG
  DEBUG ("Consumer Thread = [%lu] Stoped!\n", (unsigned long) pthread_self());
  #endif
  pthread_exit((void *) loc_hasht);
  return NULL;
}

/** @brief Function that add the key and the value to the main hashtable or update 
  * an existent value from a given local hashtables and show the word histogram.
  */
void join_show_hasht (HASHTABLE_T **loc_hasht, int num_hasht)
{
  int i, hasht_len = 0;
  HASHTABLE_T *m_hasht = NULL;
  LISTA_GENERICA_T* loc_llist = NULL;
  FREQ_WORD_T *val_loc_hasht = NULL, *val_m_hasht = NULL;
  ITERADOR_T *it = NULL;
  char *word = NULL;
  
  /**< @brief Get the total length of the local hashtables length .*/
  for (i = 0; i < num_hasht; i++)
    hasht_len += tabela_numero_elementos(loc_hasht[i]);
  
  /**< @brief The hashtable will have twice more the total of the local hashtables to support the effort rate.*/
  m_hasht = tabela_criar (hasht_len * 2, (LIBERTAR_FUNC) free_freq_word);
  
  printf ("Joining the hashtables!\n");
  
  for (i = 0; i < num_hasht; i++) {
    printf ("Processing the local hashtable %d\n", i + 1);
    
    loc_llist = tabela_criar_lista_chaves(loc_hasht[i]);
    it = lista_criar_iterador(loc_llist);
    
    while ((word = (char *) iterador_proximo_elemento (it)) != NULL && G_continue) {
      val_loc_hasht = (FREQ_WORD_T *) tabela_consultar (loc_hasht[i], word);
      /**< @brief If the key (word) do not exits in the m_hashtable, create a new key and add the value (pointer to a type struct FREQ_WORD_T).*/
      if ((val_m_hasht = (FREQ_WORD_T *) tabela_consultar (m_hasht, val_loc_hasht->word)) == NULL)
        tabela_inserir(m_hasht, val_loc_hasht->word, val_loc_hasht);
      else {
        /**< @brief If the hashkey exits update the value (f_word_aux->count) and free the struct FREQ_WORD_T pointer.*/
        val_m_hasht->count += val_loc_hasht->count;
        MY_FREE (val_loc_hasht);
        }
    }
    
    /**< @brief IF the signal is set then free the remain values of the local hashtables.*/
    if (!G_continue)
      while ((word = (char *) iterador_proximo_elemento (it)) != NULL) {
        val_loc_hasht = (FREQ_WORD_T *) tabela_consultar (loc_hasht[i], word);
        MY_FREE (val_loc_hasht);
      }
      
    iterador_destruir(&it);
    lista_destruir(&loc_llist);
    tabela_destruir(&loc_hasht[i]);
  }
    
  convert_show_histogram (m_hasht);
}
